package com.zhaotianhao.rentingservice.consumer;

import com.alibaba.fastjson.JSON;
import com.bw.springboot.stater.email.bean.MailBean;
import com.bw.springboot.stater.email.service.MailService;
import com.zhaotianhao.rentingservice.entity.Renting;
import com.zhaotianhao.rentingservice.service.IRentingService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @Author Z-7
 * @Date 2022/5/16
 */
@Slf4j
@Component
public class KafkaConsumer2 implements AcknowledgingMessageListener<String,String> {
    @Autowired
    private MailService mailSendService;
    @Autowired
    private IRentingService iRentingService;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 指定当前消费者消费的主题 以及分区
     * @param data
     * @param acknowledgment
     */
    @Override
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "month4",partitions = "1")})
    public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
        log.info("kafka消费者2执行");
        String key = data.key();
        // 防止重复消费
        if (!redisTemplate.opsForValue().setIfAbsent(key,null,1, TimeUnit.DAYS)) {
            log.info("kafka防止重复消费");
            return;
        }
        String value = data.value();
        Renting renting = JSON.parseObject(value, Renting.class);
        boolean save = iRentingService.save(renting);
        if (save) {
            // 添加mysql成功
            log.info("mysql新增完成");
            // 发送邮件到qq邮箱
            MailBean mailBean = new MailBean();
            mailBean.setSubject("新增商品租客");
            mailBean.setContent(renting.toString());
            mailBean.setRecipient("1205374753@qq.com");
            mailSendService.sendSimpleMail(mailBean);
        }
        // 删除redis备份消息
        redisTemplate.delete("message:"+key);
        // 提交偏移量
        acknowledgment.acknowledge();
    }
}
